package com.wpp.mq.controller;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * @description:
 * @author: WangPengPeng
 * @create: 2021-12-05 20:37
 **/
@Slf4j
@RestController
@RequestMapping("/api/rocketmq")
public class TagController {


    private String PRODUCER_GROUP_TAG_NAME = "producer_grp_tag_01";
    private String NAME_SRV_ADDR = "node1:9876";

    @Resource
    private DefaultMQProducer producer;

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @GetMapping("/hello")
    public String hello () {
        return "hello world";
    }

    @GetMapping("/tag")
    public String tag (@RequestParam ("topic") String topic,
                       @RequestParam (name = "tag",required = false) String tag,
                       @RequestParam("sendMessage") String sendMessage ) throws MQClientException, RemotingException, InterruptedException {
        Message message = new Message(topic,sendMessage.getBytes());

        // fixme 这里是个坑 注意 top : tag
        rocketMQTemplate.asyncSend(topic+":"+tag, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                    log.info("tagSendResult: {} , topic : {},message:{}",sendResult.getSendStatus(),topic,message);
            }

            @Override
            public void onException(Throwable e) {
                log.info("tagErrorMessage:{}",e.getMessage());
            }
        });

        return "success";
    }

    @GetMapping("/oneWay")
    public String oneWay(@RequestParam ("topic") String topic,
                         @RequestParam (name = "tag",required = false) String tag,
                         @RequestParam("sendMessage") String sendMessage ) throws MQClientException, RemotingException, InterruptedException {

        Message message = new Message("tp_tg_01", sendMessage.getBytes());
        // tag用于标记一类消息
        message.setTags("tag1");

        // keys用于建立索引的时候，hash取模将消息的索引放到SlotTable的一个Slot链表中
        message.setKeys("oid_2021-12-01_567890765");
        // oneWay
        rocketMQTemplate.sendOneWay(topic,sendMessage);


        return "success";
    }
}
